package com.bw.sparkstreaming.job1
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
/**
  * 实时的统计每个单词出现多少次
  * 数据输入：socket
  * 数据处理：sparkcore
  * 数据输出：控制台
  */
object WordCountStream {
  def main(args: Array[String]): Unit = {
    //创建程序入口，local[2]表示启动两个线程，如果是1的话，代码在执行但是看不到print出来的数据。
    //至少需要设置2个线程，因为要保证一个线程接受数据，一个处理数据。
    //如果这儿不写，默认的线程的个数，跟你的本地的电脑的cpu core的个数是一样的。
    //电脑的cpu core的个数，肯定是超过1的.
    val conf = new SparkConf().setMaster("local[2]").setAppName(this.getClass.getSimpleName)
    val ssc = new StreamingContext(conf,Seconds(2))

    //数据处理
    val dStream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.134.130",8888)
    val lines: DStream[String] = dStream.flatMap(_.split(","))
    val wordDStream: DStream[(String,Int)] = lines.map((_,1))
    val result = wordDStream.reduceByKey(_+_)

    //数据保存
    result.print()

    //启动SparkStreaming
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()
  }
}